package org.ns.learn.rpc.client.rpcinterface;

import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang.StringUtils;
import org.ns.learn.rpc.client.RpcClient;
import org.ns.learn.rpc.client.RpcClientException;
import org.ns.learn.rpc.client.RpcClientResponse;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class RpcClients {
    List<RpcClient> clients = new ArrayList<>();
    Map<String, RpcClient> rpcClientMap = new ConcurrentHashMap<>();
    AtomicInteger atindex = new AtomicInteger(0);
    Map<String, RpcResponseRecode> responseRecodeMap = new ConcurrentHashMap<>();
    String name;

    public RpcClients() {
        this.name = "rpc_client_" + RandomStringUtils.random(10, "1234567890");
    }

    public String getName() {
        return name;
    }

    public void add(RpcClient client) {
        String key = client.toString();
        RpcClient oldClient = rpcClientMap.put(key, client);
        if (oldClient != null && oldClient != client) {
            clients = new ArrayList<>(rpcClientMap.values());
            oldClient.close();
        }

    }

    public void createOrClear(List<String> hostAndPorts)  {
        Map<String, RpcClient> newClientMap = new ConcurrentHashMap<>();
        Map<String, RpcClient> oldClientMap = rpcClientMap;
        int reset = 0;
        for (int i = 0; i < hostAndPorts.size(); i++) {
            String hostAndPort = hostAndPorts.get(i);

            if (oldClientMap.containsKey(hostAndPort)) {
                newClientMap.put(hostAndPort, oldClientMap.get(hostAndPort));
            } else {
                String arrs[] = hostAndPort.split(":");
                String host = arrs[0].trim();
                int port = 2181;
                if (arrs.length == 2) {
                    port = Integer.parseInt(arrs[1].trim());
                }
                try {
                    RpcClient client = new RpcClient(host, port);
                    newClientMap.put(hostAndPort, client);
                } catch (RpcClientException e) {
                   e.printStackTrace();
                   if (reset < 10) {
                       i--;
                   }
                }
            }
            this.rpcClientMap = newClientMap;
            this.clients = new ArrayList<>(newClientMap.values());

            // 清理
            for (RpcClient client : oldClientMap.values()) {
                if (!newClientMap.containsKey(client.toString())) {
                    client.close();
                }
            }
        }

    }


    public RpcClient get() {
        List<RpcClient> clients = this.clients;
        if (clients.size() == 0) {
            return null;
        }

        RpcResponseRecode serverRecode = selectServerName();
        if (serverRecode != null) {
            String serverName = serverRecode.getClientName();
            for (RpcClient client : clients) {
                if (serverName.equals(client.toString())) {
                    System.out.println("选择rpc服务-> " + client.toString() + ", 耗时: " + serverRecode.getElapsedTime());
                    return client;
                }
            }

        }

        int index1 = atindex.incrementAndGet();
        int index2 = index1 % clients.size();
        if (index1 != index2) {
            atindex.compareAndSet(index1, index2);
        }
        RpcClient client = clients.get(index2);
        System.out.println("重选rpc服务-> " + client.toString());
        return client;
    }

    public void recodeResponseTime(RpcClientResponse response) {
        if (response == null) {
            return;
        }

        RpcClient client = response.getClient();
        String clientName = client.toString();
        RpcResponseRecode rpcResponseRecode = new RpcResponseRecode();

        rpcResponseRecode.setClientName(clientName);
        rpcResponseRecode.setRequestTime(response.getRequestTime());
        rpcResponseRecode.setResponseTime(response.getRespsonseTime());
        responseRecodeMap.put(clientName, rpcResponseRecode);
    }

    public List<RpcResponseRecode> getResponseRecodes() {
        return new ArrayList<>(responseRecodeMap.values());
    }

    public RpcResponseRecode selectServerName() {
        List<RpcResponseRecode> list = getResponseRecodes()
                .stream().filter(n->!n.isTimeout(5000)).collect(Collectors.toList());
        if (list.size() == 0 || list.size() != clients.size()) {
            return null;
        }
        List<RpcResponseRecode> selects = new ArrayList<>(list.size());

        long cmpTmp = Long.MAX_VALUE;

        for (RpcResponseRecode rpcResponseRecode : list) {

            long x = rpcResponseRecode.getElapsedTime();
            if (x < cmpTmp) {
                selects = new ArrayList<>(list.size());
                selects.add(rpcResponseRecode);
                cmpTmp = x;
            } else if (x == cmpTmp) {
                selects.add(rpcResponseRecode);
            }
        }
        if (selects.size() == 1) {
            return selects.get(0);
        } else if (selects.size() > 1) {
            return selects.get(ThreadLocalRandom.current().nextInt(selects.size()));
        } else {
            return null;
        }

    }
}
